#ifndef THREAD_POOL_H
#define THREAD_POOL_H

#include <condition_variable>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <queue>
#include <signal.h>
#include <stdexcept>
#include <thread>
#include <vector>

#include <attrs.h>
#include <httplib.h>

class ThreadPool {
public:
    ThreadPool(size_t);
    ~ThreadPool();

    template <class F>
    auto enqueue(F&& f);

private:
    /* 记录工作线程 ID，以便回收线程 */
    std::vector<std::thread> workers;
    /* 任务队列 */
    std::queue<std::string> tasks;

    /* 条件变量和互斥锁 */
    std::mutex queue_mutex;
    std::condition_variable condition;
    bool stop;
};

/* 构造函数，启动一定数量的 worker 线程 */
inline ThreadPool::ThreadPool(size_t threads) : stop(false)
{
    for (size_t i = 0; i < threads; ++i)
        workers.emplace_back(
            [this] {
                httplib::Client cli(ONOS_HOST);
                httplib::Headers headers = {{"Authorization", "Basic b25vczpyb2Nrcw=="}};

                cli.set_default_headers(headers);
                cli.set_keep_alive(true);   // 保持连接

                sigset_t set;
                int res = sigaddset(&set, SIGALRM);
                res = sigprocmask(SIG_BLOCK, &set, NULL);
                if (res != 0) {
                    perror("设置信号屏蔽失败");
                    exit(EXIT_FAILURE);
                }

                std::string task;
                for (;;) {
                    {
                        std::unique_lock<std::mutex> lock(this->queue_mutex);
                        this->condition.wait(lock,
                            [this] { return this->stop || !this->tasks.empty(); });
                        if (this->stop && this->tasks.empty())
                            return;
                        task = std::move(this->tasks.front());
                        this->tasks.pop();
                    }

                    auto res = cli.Post(ONOS_PATH, task, content_type);
                    // if (res) {
                    //     std::cout << res->status << std::endl;
                    //     std::cout << res->body << std::endl;
                    // } else {
                    if (!res) {
                        auto err = res.error();
                        std::cout << "HTTP error: " << httplib::to_string(err) << std::endl;
                    }
                }
            });
}

// add new work item to the pool
template <class F>
auto ThreadPool::enqueue(F&& f)
{
    {
        std::unique_lock<std::mutex> lock(queue_mutex);

        // don't allow enqueueing after stopping the pool
        if (stop)
            throw std::runtime_error("enqueue on stopped ThreadPool");

        tasks.emplace(f);
    }
    condition.notify_one();
    return 1;
}

// the destructor joins all threads
inline ThreadPool::~ThreadPool()
{
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        stop = true;
    }
    condition.notify_all();
    for (std::thread& worker : workers)
        worker.join();
}

#endif
